package com.sun.kafka.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.sun.kafka.entity.Message;
import com.sun.kafka.service.MessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class MessageServiceImpl implements MessageService {

    public Map<String, Map<String, String>> fieldMapping;

    @PostConstruct
    public void init() {
        // 模拟查询数据
        fieldMapping = mock();
    }

    @Override
    public void process(List<ConsumerRecord<String, String>> records) {
        for (ConsumerRecord<String, String> record : records) {
            log.info("接收消息： record = {}", record);
            String topic = record.topic();
            String value = record.value();
            if (StringUtils.isAnyBlank(topic, value)) {
                return;
            }

            Map<String, String> maps = fieldMapping.get(topic);
            if (CollectionUtils.isEmpty(maps)) {
                log.error("没有配置有效的topic映射信息!");
                return;
            }

            JSONObject jsonValue = JSONObject.parseObject(value);
            Message message = Message.builder()
                    .metric(jsonValue.getString(maps.get("metric")))
                    .time(jsonValue.getDate(maps.get("time")))
                    .tags(jsonValue.getString(maps.get("tags")))
                    .values(jsonValue.getString(maps.get("values")))
                    .build();
            log.info("convert message = {}", message);
        }
    }

    private Map<String, Map<String, String>> mock() {
        Map<String, Map<String, String>> fieldMapping = new HashMap<>();

        Map<String, String> fieldMapping1 = new HashMap<>();
        fieldMapping1.put("metric", "name");
        fieldMapping1.put("time", "ts");
        fieldMapping1.put("tags", "labels");
        fieldMapping1.put("values", "fields");
        fieldMapping.put("batch1", fieldMapping1);


        Map<String, String> fieldMapping2 = new HashMap<>();
        fieldMapping2.put("metric", "name2");
        fieldMapping2.put("time", "ts2");
        fieldMapping2.put("tags", "labels2");
        fieldMapping2.put("values", "fields2");
        fieldMapping.put("batch2", fieldMapping2);
        return fieldMapping;
    }

}
